package com.atguigu.chapter07.d_state;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Iterator;

/**
 * @ClassName: Flink01_State_Operator_List
 * @Description:
 * @Author: kele
 * @Date: 2021/4/7 19:38
 *
 * List算子的用法示例
 *
 *
 * 在map算子中计算数据的个数
 **/
public class Flink01_State_Operator_List {

    public static void main(String[] args) {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port",20000);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(2);

        //设置周期自动提交时间
        env.enableCheckpointing(1000);

        env.socketTextStream("hadoop162",8888)
                .map(new MyMapFunction())
                .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    public static class MyMapFunction implements MapFunction<String,String>, CheckpointedFunction{

        private ListState<Long> countstate;
        Long count = 0l;

        @Override
        public String map(String value) throws Exception {

            count ++;
            //计算完需要更新算子的状态
            countstate.clear();
            countstate.add(count);

            return ""+count;
        }

        /**
         * 用来初始化，当初始化的时候调用这个方法
         * @param context
         * @throws Exception
         */
        @Override
        public void initializeState(FunctionInitializationContext context) throws Exception {

            System.out.println("init.....");
/***list***/
            /**
             * 获取算子的状态：
             *
             * 状态恢复的时候 平均分配到各个并行度中
             */
          //  countstate = context.getOperatorStateStore().getListState(new ListStateDescriptor<Long>("countstate", long.class));
/***unionlist***/
            /**
             * 获取联合状态列表
             *
             * 状态恢复的时候将所有数据整合之后分别分配到各个并行度中
             */

            countstate = context.getOperatorStateStore().getUnionListState(new ListStateDescriptor<Long>("counstate",long.class));


            //初始化，（1、刚开始，2、断电后继续）如果countstate中有值，将countstate中的值赋值给count
            Iterator<Long> it = countstate.get().iterator();

            if( it.hasNext() ){
                count = it.next();
            }
        }

        /**
         * 保存状态，按周期
         * @param context
         * @throws Exception
         */
        @Override
        public void snapshotState(FunctionSnapshotContext context) throws Exception {

            System.out.println("save cycle");
            //周期性的保存算子的状态
            countstate.clear();
            countstate.add(count);
        }
    }


}
